package io.github.talelin.latticy.task;

import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import io.github.talelin.latticy.common.constant.Constant;
import io.github.talelin.latticy.model.MsgLogDO;
import io.github.talelin.latticy.mq.MessageHelper;
import io.github.talelin.latticy.service.MsgLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

@Component
@Slf4j
public class ResendMsg {

    @Autowired
    private MsgLogService msgLogService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 最大投递次数
    private static final int MAX_TRY_COUNT = 3;

    /**
     * 每30s拉取投递失败的消息, 重新投递
     */
    @Scheduled(cron = "0/30 * * * * ?")
    public void resend() {
        log.info("开始执行定时任务(重新投递消息)");
        QueryWrapper<MsgLogDO> wrapper=new QueryWrapper<>();
        SimpleDateFormat dateFormat=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        wrapper.lambda().eq(MsgLogDO::getStatus, Constant.MsgLogStatus.DELIVERING).le(MsgLogDO::getNextTryTime, dateFormat.format(new Date()));
        List<MsgLogDO> msgLogs = msgLogService.list(wrapper);
        msgLogs.forEach(msgLog -> {
            Long msgId = msgLog.getMsgId();
            if (msgLog.getTryCount() >= MAX_TRY_COUNT) {
                MsgLogDO msgLogDO=new MsgLogDO();
                msgLogDO.setMsgId(msgId);
                msgLogDO.setStatus(Constant.MsgLogStatus.DELIVER_FAIL);
                msgLogService.updateById(msgLogDO);
                log.info("超过最大重试次数, 消息投递失败, msgId: {}", msgId);
            } else {
                UpdateWrapper updateWrapper = new UpdateWrapper();
                updateWrapper.eq("msg_id",msgId);
                updateWrapper.setSql("try_count = try_count + 1");
                msgLogService.update(updateWrapper);// 投递次数+1
                CorrelationData correlationData = new CorrelationData(String.valueOf(msgId));
                rabbitTemplate.convertAndSend(msgLog.getExchange(), msgLog.getRoutingKey(), MessageHelper.objToMsg(msgLog.getMsg()), correlationData);// 重新投递

                log.info("第 " + (msgLog.getTryCount() + 1) + " 次重新投递消息");
            }
        });

        log.info("定时任务执行结束(重新投递消息)");
    }

}
